package com.suse.service.community.kafka;

import com.alibaba.fastjson2.JSONObject;
import com.suse.service.community.constant.TopicConstant;
import com.suse.service.community.constant.UserConstant;
import com.suse.service.community.entity.DiscussPost;
import com.suse.service.community.entity.Message;
import com.suse.service.community.service.DiscussPostService;
import com.suse.service.community.service.ElasticsearchService;
import com.suse.service.community.service.MessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

/**
 * <p>
 * 事件消费者
 * </p>
 *
 * @author 辜玉彬
 * @since 2022-09-03 15:14
 */
@Slf4j
@Component
public class EventConsumer {

    @Resource
    private MessageService messageService;

    @Resource
    private DiscussPostService discussPostService;

    @Resource
    private ElasticsearchService elasticsearchService;

    @KafkaListener(topics = {TopicConstant.COMMENT,TopicConstant.LIKE,TopicConstant.FOLLOW})
    public void handleEvent(ConsumerRecord<String,Object> record) {
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        Map<String, Object> content = event.getContent();
        Message message = new Message();
        // 发送站内通知
        message.setFromId(UserConstant.SYSTEM_USER_ID);
        message.setToId((Integer) content.get("entityUserId"));
        message.setConversationId(event.getTopic());
        message.setContent(JSONObject.toJSONString(content));
        messageService.save(message);
    }

    @KafkaListener(topics = {TopicConstant.PUBLISH})
    public void handlePublish(ConsumerRecord<String,Object> record) {
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        int postId = (Integer) event.get("postId");
        DiscussPost post = discussPostService.getById(postId);
        elasticsearchService.saveDiscussPost(post);
    }

    @KafkaListener(topics = {TopicConstant.POST_DELETE})
    public void handlePostDelete(ConsumerRecord<String,Object> record) {
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        int postId = (Integer) event.get("postId");
        elasticsearchService.removeDiscussPost(postId);
    }

}
